package com.codejiwei.flink.sql;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableDescriptor;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.time.LocalDateTime;

/**
 * author: codejiwei
 * date: 2023/8/16
 * desc: flink sql regular join
 **/
public class Flink_SQL_RegularJoin {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //dim table
        SingleOutputStreamOperator<Row> dimDS = env.fromElements(
                Row.of(1, "Tom", 22, LocalDateTime.parse("2023-08-15T16:39:50.076")),
                Row.of(2, "Jerry", 33, LocalDateTime.parse("2023-08-15T16:40:51.076")),
                Row.of(3, "Marry", 44, LocalDateTime.parse("2023-08-15T16:43:52.076"))
        ).returns(Types.ROW_NAMED(new String[]{"id", "name", "age", "ts"}, Types.INT, Types.STRING, Types.INT, Types.LOCAL_DATE_TIME));
        tableEnv.createTemporaryView("dimTable", dimDS);

        //source table
        Table table = tableEnv.from(TableDescriptor.forConnector("datagen")
                .schema(Schema.newBuilder()
                        .column("id", DataTypes.INT().notNull())
                        .column("sensor_id", DataTypes.STRING())
                        .column("event_time", DataTypes.TIMESTAMP(3))
                        .primaryKey("id")
                        .watermark("event_time", "event_time - INTERVAL '3' SECOND")
                        .build())
                .option("rows-per-second", "1")
//                        .option("number-of-rows", "10")
                .option("fields.sensor_id.length", "5")
                .option("fields.id.min", "0")
                .option("fields.id.max", "10")
                .build());

        tableEnv.createTemporaryView("sourceTable", table);

        //TODO left/right/full out join/inner join
//        Table table1 = tableEnv.sqlQuery("select * from sourceTable left join dimTable on sourceTable.id = dimTable.id");

        //TODO interval join
//        Table table1 = tableEnv.sqlQuery("select * from sourceTable st, dimTable dt where st.id = dt.id and st.event_time between dt.ts - interval '10' MINUTE and dt.ts");

        Table table1 = tableEnv.sqlQuery("select * from sourceTable as st join dimTable for SYSTEM_TIME as of st.event_time as e on st.id = e.id");

//        tableEnv.toDataStream(table1).print();
        tableEnv.toChangelogStream(table1).print();

        env.execute();
    }
}
